﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.WindowsAzure.StorageClient;
using System.Reactive;
using System.Reactive.Linq;
using System.Diagnostics.Contracts;

namespace Microsoft.WindowsAzure.StorageClient
{
    /// <summary>
    /// Extension methods for CloudQueue.
    /// </summary>
    public static class CloudQueueExtensions
    {
        /// <summary>
        /// Deletes message asynchronous.
        /// </summary>
        /// <param name="queue">The cloud queue.</param>
        /// <param name="message">The message.</param>
        public static IObservable<Unit> DeleteMessageAsync(this CloudQueue queue, CloudQueueMessage message)
        {
            Contract.Requires(queue != null);
            Contract.Requires(message != null);
            Contract.Ensures(Contract.Result<IObservable<Unit>>() != null);

            var deleteMessage = Observable.FromAsyncPattern<CloudQueueMessage>(
                                                queue.BeginDeleteMessage,
                                                queue.EndDeleteMessage);

            return Observable.Defer(() => deleteMessage(message));
        }

        /// <summary>
        /// Adds message asynchronous.
        /// </summary>
        /// <param name="queue">The cloud queue.</param>
        /// <param name="message">The message.</param>
        public static IObservable<Unit> AddMessageAsync(this CloudQueue queue, CloudQueueMessage message)
        {
            Contract.Requires(queue != null);
            Contract.Requires(message != null);
            Contract.Ensures(Contract.Result<IObservable<Unit>>() != null);

            var addMessage = Observable.FromAsyncPattern<CloudQueueMessage>(
                                                queue.BeginAddMessage,
                                                queue.EndAddMessage);

            return Observable.Defer(() => addMessage(message));
        }

        /// <summary>
        /// Get single message asynchronous.
        /// </summary>
        /// <param name="queue">Cloud queue.</param>
        public static IObservable<CloudQueueMessage> GetMessageAsync(this CloudQueue queue)
        {
            Contract.Requires(queue != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            var getMessage = Observable.FromAsyncPattern<CloudQueueMessage>(
                                                queue.BeginGetMessage,
                                                queue.EndGetMessage);

            return Observable.Defer(getMessage);
        }

        /// <summary>
        /// Get multiple messages asynchronous.
        /// </summary>
        /// <param name="queue">The cloud queue.</param>
        /// <param name="messageCount">The number of messages to retrieve.</param>
        public static IObservable<IEnumerable<CloudQueueMessage>> GetMessagesAsync(this CloudQueue queue, int messageCount)
        {
            Contract.Requires(queue != null);
            Contract.Ensures(Contract.Result<IObservable<CloudQueueMessage>>() != null);

            var getMessages = Observable.FromAsyncPattern<int, IEnumerable<CloudQueueMessage>>(
                                                queue.BeginGetMessages,
                                                queue.EndGetMessages);

            return Observable.Defer(() => getMessages(messageCount));
        }
    }
}
